package com.ma.chapter4.t4;

import lombok.extern.slf4j.Slf4j;

import java.util.LinkedList;
import java.util.List;

/**
 * @ClassName TestProductConsumer
 * @Author: mayongqiang
 * @DATE 2022/3/10 20:15
 * @Description: 生产者-消费者模式
 */
@Slf4j(topic = "m.test")
public class TestProductConsumer {


    public static void main(String[] args) {
        MessageQueue messageQueue = new MessageQueue(2);
        for (int i = 0; i < 4; i++) {
            int id = i;
            new Thread(() -> {
                log.debug("download");
                try {
                    List<String> download = Downloader.download();
                    log.debug("try put message ({})", id);
                    messageQueue.put(new Message(id, download));
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }, "生产者" + i).start();
        }


        new Thread(() -> {
            while (true) {
                try {
                    Message message = messageQueue.take();
                    List<String> response = (List<String>) message.getMessage();
                    log.debug("take message({}): [{}] lines", message.getId(), response.size());
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

            }
        }, "消费者").start();
    }

}

@Slf4j(topic = "m.Message")
class Message {
    private int id;
    private Object message;

    public Message(int id, Object message) {
        this.id = id;
        this.message = message;
    }

    public Object getMessage() {
        return message;
    }

    public int getId() {
        return id;
    }
}

@Slf4j(topic = "m.MessageQueue")
class MessageQueue {
    private LinkedList<Message> queue;
    private int capacity;

    public MessageQueue(int capacity) {
        this.capacity = capacity;
        queue = new LinkedList<>();
    }

    public Message take() throws InterruptedException {
        synchronized (queue) {
            while (queue.isEmpty()) {
                log.debug("没货了... wait");
                queue.wait();
            }
            Message message = queue.removeFirst();
            queue.notifyAll();
            return message;
        }
    }

    public void put(Message message) throws InterruptedException {
        synchronized (queue) {
            while (queue.size() == capacity) {
                log.debug("库存已达上限...wait");
                queue.wait();
            }
            queue.addLast(message);
            queue.notifyAll();
        }
    }
}